//! Redis数据源实现
//! 
//! 基于redis-rs的Redis数据源插件

use std::collections::HashMap;
use async_trait::async_trait;
use redis::{Client, Connection, Commands, RedisResult, Value as RedisValue};
use serde::{Serialize, Deserialize};
use serde_json::Value;

use crate::{DataSource, DataSourceQuery, DataSourceResult, DataSourceStats};

/// Redis数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RedisConfig {
    /// Redis连接URL
    pub url: String,
    /// 数据库编号
    pub database: u8,
    /// 连接超时时间（秒）
    pub connect_timeout: u64,
    /// 命令超时时间（秒）
    pub command_timeout: u64,
}

impl Default for RedisConfig {
    fn default() -> Self {
        Self {
            url: "redis://localhost:6379".to_string(),
            database: 0,
            connect_timeout: 30,
            command_timeout: 10,
        }
    }
}

/// Redis数据源
pub struct RedisDataSource {
    config: RedisConfig,
    client: Option<Client>,
    connection: Option<Connection>,
}

impl RedisDataSource {
    /// 创建新的Redis数据源
    pub fn new(config: RedisConfig) -> Self {
        Self {
            config,
            client: None,
            connection: None,
        }
    }
    
    /// 获取Redis连接
    fn get_connection(&mut self) -> anyhow::Result<&mut Connection> {
        self.connection.as_mut().ok_or_else(|| {
            anyhow::anyhow!("Redis未连接")
        })
    }
    
    /// 将Redis值转换为JSON值
    fn redis_value_to_json(value: RedisValue) -> Value {
        match value {
            RedisValue::Nil => Value::Null,
            RedisValue::Int(i) => Value::Number(i.into()),
            RedisValue::Data(bytes) => {
                match String::from_utf8(bytes) {
                    Ok(s) => {
                        // 尝试解析为JSON
                        serde_json::from_str(&s).unwrap_or(Value::String(s))
                    }
                    Err(_) => Value::Null,
                }
            }
            RedisValue::Bulk(values) => {
                Value::Array(values.into_iter().map(Self::redis_value_to_json).collect())
            }
            RedisValue::Status(s) => Value::String(s),
            RedisValue::Okay => Value::String("OK".to_string()),
        }
    }
    
    /// 执行Redis命令
    async fn execute_redis_command(&mut self, command: &str, args: &[String]) -> anyhow::Result<Value> {
        let conn = self.get_connection()?;
        
        match command.to_uppercase().as_str() {
            "GET" => {
                if args.is_empty() {
                    return Err(anyhow::anyhow!("GET命令需要key参数"));
                }
                let result: RedisResult<RedisValue> = conn.get(&args[0]);
                match result {
                    Ok(value) => Ok(Self::redis_value_to_json(value)),
                    Err(e) => Err(anyhow::anyhow!("GET命令执行失败: {}", e)),
                }
            }
            "SET" => {
                if args.len() < 2 {
                    return Err(anyhow::anyhow!("SET命令需要key和value参数"));
                }
                let result: RedisResult<RedisValue> = conn.set(&args[0], &args[1]);
                match result {
                    Ok(value) => Ok(Self::redis_value_to_json(value)),
                    Err(e) => Err(anyhow::anyhow!("SET命令执行失败: {}", e)),
                }
            }
            "DEL" => {
                if args.is_empty() {
                    return Err(anyhow::anyhow!("DEL命令需要key参数"));
                }
                let result: RedisResult<i32> = conn.del(&args[0]);
                match result {
                    Ok(count) => Ok(Value::Number(count.into())),
                    Err(e) => Err(anyhow::anyhow!("DEL命令执行失败: {}", e)),
                }
            }
            "EXISTS" => {
                if args.is_empty() {
                    return Err(anyhow::anyhow!("EXISTS命令需要key参数"));
                }
                let result: RedisResult<bool> = conn.exists(&args[0]);
                match result {
                    Ok(exists) => Ok(Value::Bool(exists)),
                    Err(e) => Err(anyhow::anyhow!("EXISTS命令执行失败: {}", e)),
                }
            }
            "KEYS" => {
                let pattern = args.get(0).map(|s| s.as_str()).unwrap_or("*");
                let result: RedisResult<Vec<String>> = conn.keys(pattern);
                match result {
                    Ok(keys) => Ok(Value::Array(keys.into_iter().map(Value::String).collect())),
                    Err(e) => Err(anyhow::anyhow!("KEYS命令执行失败: {}", e)),
                }
            }
            "HGET" => {
                if args.len() < 2 {
                    return Err(anyhow::anyhow!("HGET命令需要key和field参数"));
                }
                let result: RedisResult<RedisValue> = conn.hget(&args[0], &args[1]);
                match result {
                    Ok(value) => Ok(Self::redis_value_to_json(value)),
                    Err(e) => Err(anyhow::anyhow!("HGET命令执行失败: {}", e)),
                }
            }
            "HSET" => {
                if args.len() < 3 {
                    return Err(anyhow::anyhow!("HSET命令需要key、field和value参数"));
                }
                let result: RedisResult<i32> = conn.hset(&args[0], &args[1], &args[2]);
                match result {
                    Ok(count) => Ok(Value::Number(count.into())),
                    Err(e) => Err(anyhow::anyhow!("HSET命令执行失败: {}", e)),
                }
            }
            "HGETALL" => {
                if args.is_empty() {
                    return Err(anyhow::anyhow!("HGETALL命令需要key参数"));
                }
                let result: RedisResult<HashMap<String, String>> = conn.hgetall(&args[0]);
                match result {
                    Ok(hash) => {
                        let map: serde_json::Map<String, Value> = hash.into_iter()
                            .map(|(k, v)| (k, Value::String(v)))
                            .collect();
                        Ok(Value::Object(map))
                    }
                    Err(e) => Err(anyhow::anyhow!("HGETALL命令执行失败: {}", e)),
                }
            }
            "PING" => {
                let result: RedisResult<String> = redis::cmd("PING").query(conn);
                match result {
                    Ok(response) => Ok(Value::String(response)),
                    Err(e) => Err(anyhow::anyhow!("PING命令执行失败: {}", e)),
                }
            }
            _ => Err(anyhow::anyhow!("不支持的Redis命令: {}", command)),
        }
    }
    
    /// 解析查询内容为Redis命令
    fn parse_query(content: &str) -> anyhow::Result<(String, Vec<String>)> {
        let parts: Vec<&str> = content.trim().split_whitespace().collect();
        if parts.is_empty() {
            return Err(anyhow::anyhow!("查询内容为空"));
        }
        
        let command = parts[0].to_string();
        let args = parts[1..].iter().map(|s| s.to_string()).collect();
        
        Ok((command, args))
    }
}

#[async_trait]
impl DataSource for RedisDataSource {
    fn name(&self) -> &str {
        "Redis"
    }
    
    fn description(&self) -> &str {
        "Redis内存数据库数据源"
    }
    
    async fn connect(&mut self) -> anyhow::Result<()> {
        let client = Client::open(self.config.url.as_str())
            .map_err(|e| anyhow::anyhow!("创建Redis客户端失败: {}", e))?;
            
        let mut connection = client.get_connection()
            .map_err(|e| anyhow::anyhow!("连接Redis失败: {}", e))?;
            
        // 选择数据库
        if self.config.database != 0 {
            let _: () = redis::cmd("SELECT")
                .arg(self.config.database)
                .query(&mut connection)
                .map_err(|e| anyhow::anyhow!("选择Redis数据库失败: {}", e))?;
        }
        
        self.client = Some(client);
        self.connection = Some(connection);
        Ok(())
    }
    
    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.connection = None;
        self.client = None;
        Ok(())
    }
    
    async fn is_connected(&self) -> bool {
        self.connection.is_some()
    }
    
    async fn execute_query(&mut self, query: DataSourceQuery) -> anyhow::Result<DataSourceResult> {
        let start_time = std::time::Instant::now();
        
        // 解析查询
        let (command, args) = Self::parse_query(&query.content)?;
        
        // 执行Redis命令
        match self.execute_redis_command(&command, &args).await {
            Ok(result) => {
                let duration = start_time.elapsed();
                
                Ok(DataSourceResult {
                    success: true,
                    data: result,
                    error: None,
                    stats: DataSourceStats {
                        execution_time_ms: duration.as_millis() as f64,
                        rows_processed: 1, // Redis命令通常处理一个操作
                        bytes_processed: 0, // TODO: 计算实际字节数
                    },
                })
            }
            Err(e) => {
                let duration = start_time.elapsed();
                
                Ok(DataSourceResult {
                    success: false,
                    data: Value::Null,
                    error: Some(e.to_string()),
                    stats: DataSourceStats {
                        execution_time_ms: duration.as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                })
            }
        }
    }
    
    async fn health_check(&self) -> anyhow::Result<Value> {
        if !self.is_connected().await {
            return Ok(serde_json::json!({
                "status": "unhealthy",
                "database": "Redis",
                "connected": false,
                "error": "未连接"
            }));
        }
        
        // 这里我们不能直接使用连接，因为health_check是&self方法
        // 在实际应用中，可能需要重新设计这个方法或使用Arc<Mutex<Connection>>
        Ok(serde_json::json!({
            "status": "healthy",
            "database": "Redis",
            "connected": true,
            "database_number": self.config.database
        }))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_redis_data_source_creation() {
        let config = RedisConfig::default();
        let data_source = RedisDataSource::new(config);
        
        assert_eq!(data_source.name(), "Redis");
        assert_eq!(data_source.description(), "Redis内存数据库数据源");
        assert!(!data_source.is_connected().await);
    }

    #[tokio::test]
    async fn test_redis_config_default() {
        let config = RedisConfig::default();
        assert_eq!(config.url, "redis://localhost:6379");
        assert_eq!(config.database, 0);
        assert_eq!(config.connect_timeout, 30);
        assert_eq!(config.command_timeout, 10);
    }

    #[test]
    fn test_parse_query() {
        let (command, args) = RedisDataSource::parse_query("GET mykey").unwrap();
        assert_eq!(command, "GET");
        assert_eq!(args, vec!["mykey"]);
        
        let (command, args) = RedisDataSource::parse_query("SET mykey myvalue").unwrap();
        assert_eq!(command, "SET");
        assert_eq!(args, vec!["mykey", "myvalue"]);
        
        let (command, args) = RedisDataSource::parse_query("HSET myhash field value").unwrap();
        assert_eq!(command, "HSET");
        assert_eq!(args, vec!["myhash", "field", "value"]);
    }

    #[test]
    fn test_redis_value_to_json() {
        assert_eq!(RedisDataSource::redis_value_to_json(RedisValue::Nil), Value::Null);
        assert_eq!(RedisDataSource::redis_value_to_json(RedisValue::Int(42)), Value::Number(42.into()));
        assert_eq!(RedisDataSource::redis_value_to_json(RedisValue::Status("OK".to_string())), Value::String("OK".to_string()));
        assert_eq!(RedisDataSource::redis_value_to_json(RedisValue::Okay), Value::String("OK".to_string()));
    }
}
